/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2018-2022. All rights reserved.
 * Description: HdcsImpl接口实现文件
 * Author: liutianming
 * Create: 2018-01-05
 */

#include "hccl_ex.h"
#include "log.h"
#include "transport_shm_event_pub.h"
#include "dlhdds_function.h"
#include "adapter_hdds.h"
#include "hccd_private.h"
#include "aicpu_sharder.h"
#include "hdcs_impl.h"
#include "../../../hccl_heterog/rpc/hcom_rpc.h"

namespace hccl {
using namespace std;

HdcsImpl::HdcsImpl() : initFlag_(false), comm_(nullptr), embeddingRequestPool_(RING_MEMORY_CAPACITY)
{
}

HdcsImpl::~HdcsImpl()
{
    HCCL_INFO("[HdcsImpl]start hdds destory");
    DeInit();
}

HcclResult HdcsImpl::Init(HdcsCsInitPara para, std::vector<u32> psRankId)
{
    if (initFlag_) {
        return HCCL_SUCCESS;
    }

    if (para.isInference) {
        HCCL_DEBUG("[HdcsImpl]HdcsImpl::Init isInference ");
        initFlag_ = true;
        return HCCL_SUCCESS;
    }

    CHK_RET(DlHddsFunction::GetInstance().DlHddsFunctionInit());

    CommAttr attr;
    attr.mode = HCCL_MODE_AI_CPU;
    HcomInfo rankInfo;
    CHK_RET(HcclRPCInitComm(para.rankTableM, para.rank, &attr, &comm_, rankInfo));
    CHK_PTR_NULL(comm_);

    u32 partitionNum = 65537;
    u32 psNum = para.psSize;
    u32 psIds[psNum];
    u32 pos = 0;
    for (u32 psId = 0; psId < psNum; psId++) {
        psIds[pos] = psId;
        pos++;
    }

    HCCL_RUN_INFO("[Init][HddsImpl]hrtInitPartitionMap partitionNum[%u] psNum[%u].", partitionNum, psNum);
    CHK_RET(hrtInitPartitionMap(partitionNum, psNum, psIds));
    // 这里初始化传入psRankId
    CHK_RET(opRemoteLookUpExecutor_.Init(psRankId));
    CHK_RET(opRemoteUpdateExecutor_.Init(psRankId));
    CHK_RET(embeddingRequestPool_.Init());
    psRankId_ = psRankId;
    initFlag_ = true;
    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::DeInit()
{
    initFlag_ = false;
    psRankId_.clear();

    auto freeRequest = [this] (unordered_map<s32, HcclOperator<EmbeddingServiceParam>*> &paraMap) -> void {
        for (auto &iter : paraMap) {
            if (iter.second != nullptr) {
                embeddingRequestPool_.Free(iter.second);
                iter.second = nullptr;
            }
        }
    };

    EXECEPTION_CATCH(lookupParamsMap_.EraseAll(freeRequest), return HCCL_E_INTERNAL);

    auto updateFreeRequest = [this] (HcclOperator<EmbeddingServiceParam> *&opPtr) -> void {
        if (opPtr != nullptr) {
            embeddingRequestPool_.Free(opPtr);
            opPtr = nullptr;
        }
    };

    EXECEPTION_CATCH(updateParamsMap_.EraseAll(updateFreeRequest), return HCCL_E_INTERNAL);

    if (comm_ != nullptr) {
        return HcclFinalizeCsComm(comm_);
    }
    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::RegTransport(HdcsRegTransportPara para)
{
    vector<TransportShmEventMember> transportPara(para.pcieTransportNum);
    s32 pcieTransCnt = 0;

    for (s32 i = 0; i < para.transportNum; i++) {
        HCCL_DEBUG("pcieTransCnt[%d], transportInfos[%d].commType = %d", pcieTransCnt, i,
            para.transportInfos[i].commType);

        if (para.transportInfos[i].commType != HcclHeterogCommType::PCIE) {
            continue;
        }

        if (pcieTransCnt >= para.pcieTransportNum) {
            HCCL_ERROR("failed, pcieTransCnt[%d] >= pcieTransportNum[%d]", pcieTransCnt, para.pcieTransportNum);
            return HCCL_E_INTERNAL;
        }

        TransportShmEventMember &transportInfo = transportPara[pcieTransCnt];

        transportInfo.deviceLogicId = para.transportInfos[i].deviceLogicId;
        transportInfo.rank = para.transportInfos[i].rank;
        transportInfo.tag = para.transportInfos[i].tag;
        transportInfo.inputShmMem = DeviceMem::create(para.transportInfos[i].inputShmMem,
            para.transportInfos[i].inputMemSize);
        transportInfo.outputShmMem = DeviceMem::create(para.transportInfos[i].outputShmMem,
            para.transportInfos[i].outputMemSize);
        transportInfo.envelopeShmQue = DeviceMem::create(para.transportInfos[i].envelopMem,
            para.transportInfos[i].envelopeSize);
        pcieTransCnt++;
    }

    void *transportParaPtr = transportPara.data();
    CHK_RET(RegTransportLinks(pcieTransCnt, transportParaPtr, comm_));

    return HCCL_SUCCESS;
}

bool HdcsImpl::IsInited()
{
    return initFlag_;
}

HcclResult HdcsImpl::RemoteLookUp(HdcsEmbeddingServiceParam *lookupInfo)
{
    u64 keySize = lookupInfo->keyMaxNum * SIZE_TABLE[lookupInfo->keyType];
    u64 valueSize = lookupInfo->keyMaxNum * lookupInfo->valueItemSize;
    if (lookupInfo->keyTransferMemSize < keySize || lookupInfo->valueTransferMemSize < valueSize) {
        HCCL_ERROR("TransferMemSize not enough!!! keyTransferMemSize[%lld] valueTransferMemSize[%lld]" \
            " keySize[%lld] valueSize[%lld]", lookupInfo->keyTransferMemSize, lookupInfo->valueTransferMemSize,
            keySize, valueSize);
        return HCCL_E_MEMORY;
    }

    HcclOperator<EmbeddingServiceParam> *ptr = embeddingRequestPool_.Alloc();
    CHK_PTR_NULL(ptr);
    EmbeddingServiceParam &param = ptr->param_;
    param = *lookupInfo;
    param.opType = EmbeddingServiceParam::OperatorType::REMOTE_LOOK_UP;
    param.embeddingParam.comm = comm_;

    CHK_RET(lookupAlg_.Process(ptr));
    CHK_RET(opRemoteLookUpExecutor_.ExecuteOp(ptr));

    embeddingRequestPool_.Free(ptr);
    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::RemoteUpdate(HdcsEmbeddingServiceParam *updateInfo)
{
    u64 keySize = updateInfo->keyMaxNum * SIZE_TABLE[updateInfo->keyType];
    u64 valueSize = updateInfo->keyMaxNum * updateInfo->valueItemSize;
    if (updateInfo->keyTransferMemSize < keySize || updateInfo->valueTransferMemSize < valueSize) {
        HCCL_ERROR("TransferMemSize not enough!!! keyTransferMemSize[%lld] valueTransferMemSize[%lld] keySize[%lld]" \
            " valueSize[%lld]", updateInfo->keyTransferMemSize, updateInfo->valueTransferMemSize, keySize, valueSize);
        return HCCL_E_MEMORY;
    }

    HcclOperator<EmbeddingServiceParam> *ptr = embeddingRequestPool_.Alloc();
    CHK_PTR_NULL(ptr);
    EmbeddingServiceParam &param = ptr->param_;
    param = *updateInfo;
    param.opType = EmbeddingServiceParam::OperatorType::REMOTE_UPDATE;
    param.embeddingParam.comm = comm_;

    HCCL_DEBUG("RemoteUpdate: key[%llu] tag[%d]", *static_cast<u64 *>(updateInfo->keys), param.tag);

    CHK_RET(updateAlg_.Process(ptr));
    CHK_RET(opRemoteUpdateExecutor_.ExecuteOp(ptr));
    embeddingRequestPool_.Free(ptr);
    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::IrecvLookupResponse(HcclRequest request)
{
    HcclOpPtr ptr = static_cast<HcclOpPtr>(request);
    HcclOperator<EmbeddingServiceParam> *paraPtr = static_cast<HcclOperator<EmbeddingServiceParam> *>(request);
    paraPtr->param_.opType = EmbeddingServiceParam::OperatorType::WAIT_LOOK_UP_DATA;
    paraPtr->param_.embeddingParam.comm = comm_;

    CHK_RET(lookupAlg_.Process(ptr));
    CHK_RET(opRemoteLookUpExecutor_.ExecuteOp(ptr));
    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::IrecvUpdateResponse(HcclRequest request)
{
    HcclOpPtr ptr = static_cast<HcclOpPtr>(request);
    HcclOperator<EmbeddingServiceParam> *paraPtr = static_cast<HcclOperator<EmbeddingServiceParam> *>(request);
    paraPtr->param_.opType = EmbeddingServiceParam::OperatorType::WAIT_UPDATE_DATA;

    CHK_RET(updateAlg_.Process(ptr));
    CHK_RET(opRemoteLookUpExecutor_.ExecuteOp(ptr));
    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::TestSome(HcclRequest* request)
{
    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::CollLookupKeysDuplicates(HdcsEmbeddingServiceParam *lookupInfo)
{
    CHK_PTR_NULL(lookupInfo);
    CHK_PTR_NULL(lookupInfo->keyTransferMem);
    CHK_PTR_NULL(lookupInfo->valueTransferMem);
    CHK_PTR_NULL(lookupInfo->keys);
    CHK_PTR_NULL(lookupInfo->values);

    if (lookupInfo->pairedMode) {
        CHK_PTR_NULL(lookupInfo->indices);
        CHK_PTR_NULL(lookupInfo->numUniqued);
        CHK_PTR_NULL(lookupInfo->psSeg);
        CHK_PTR_NULL(lookupInfo->psSegNum);
    }

    CHK_PRT_RET(lookupInfo->keyType >= HCCL_DATA_TYPE_RESERVED, HCCL_ERROR("keyType Invalid"), HCCL_E_PARA);
    u64 keySize = lookupInfo->keyMaxNum * SIZE_TABLE[lookupInfo->keyType];
    u64 valueSize = lookupInfo->keyMaxNum * lookupInfo->valueItemSize;
    if (lookupInfo->keyTransferMemSize < keySize || lookupInfo->valueTransferMemSize < valueSize) {
        HCCL_ERROR("TransferMemSize not enough! keyTransferMemSize[%lld] valueTransferMemSize[%lld]" \
            " keySize[%lld] valueSize[%lld]", lookupInfo->keyTransferMemSize, lookupInfo->valueTransferMemSize,
            keySize, valueSize);
        return HCCL_E_MEMORY;
    }

    s32 tag = lookupInfo->tag;
    s32 cubeIndex = lookupInfo->cubeIndex;

    auto allocRequest = [this] (unordered_map<s32, HcclOperator<EmbeddingServiceParam>*> &paraMap, s32 cubeIndex) ->
        void {
        if (paraMap.find(cubeIndex) == paraMap.end()) {
            paraMap[cubeIndex] = embeddingRequestPool_.Alloc();
        }
    };

    pair<UniversalConcurrentMap<s32, unordered_map<s32, HcclOperator<EmbeddingServiceParam>*>>::Iterator, bool> it;
    EXECEPTION_CATCH(it = lookupParamsMap_.EmplaceAndUpdate(tag, allocRequest, cubeIndex), return HCCL_E_INTERNAL);
    if (it.second) {
        HCCL_DEBUG("lookupParamsMap_ new insert, tag[%d]", tag);
    }

    HcclOperator<EmbeddingServiceParam> *ptr = it.first->second[cubeIndex];
    CHK_PTR_NULL(ptr);

    // 下面执行时捕获异常
    auto exeAlloc = [this] (unordered_map<s32, unique_ptr<OpExecutor>> &map, s32 cubeIndex) -> void {
        if (map.find(cubeIndex) == map.end()) {
            map[cubeIndex] = make_unique<HdcsOpRemoteLookUpExecutor>();
        }
    };

    pair<UniversalConcurrentMap<s32, unordered_map<s32, unique_ptr<OpExecutor>>>::Iterator, bool> exeIt;
    EXECEPTION_CATCH(exeIt = lookupOpExecutorMap_.EmplaceAndUpdate(tag, exeAlloc, cubeIndex), return HCCL_E_INTERNAL);

    OpExecutor *excutor = exeIt.first->second[cubeIndex].get();
    CHK_PTR_NULL(excutor);

    if (exeIt.second) {
        HdcsOpRemoteLookUpExecutor *lookExcutor = dynamic_cast<HdcsOpRemoteLookUpExecutor *>(excutor);
        CHK_PTR_NULL(lookExcutor);

        CHK_RET(lookExcutor->Init(psRankId_, lookupInfo->pipelineKeyNum));
        HCCL_DEBUG("lookupOpExecutorMap_ new insert, tag[%d]", tag);
    }

    // 当前相同tag不存在并发，参数赋值可不用加锁
    EmbeddingServiceParam &param = ptr->param_;
    param = *lookupInfo;
    param.opType = EmbeddingServiceParam::OperatorType::REMOTE_LOOKUP_KEYS_DUPLICATES;
    // comm是用来调用hccd接口的
    param.embeddingParam.comm = comm_;

    HCCL_DEBUG("CollLookupKeysDuplicates rdmaEnveInfosTransferMem[%llu], rdmaEnveInfosTransferMemSize[%llu], "
        "haveRdmaConn[%d]", hash<void *>{}(param.embeddingParam.rdmaEnveInfosTransferMem),
        lookupInfo->rdmaEnveInfosTransferMemSize, param.haveRdmaConn);

    if (param.pairedMode) {
        param.indices = lookupInfo->indices;
        param.numUniqued = lookupInfo->numUniqued;
        param.psSeg = lookupInfo->psSeg;
        param.psSegNum = lookupInfo->psSegNum;
    }

    CHK_RET(lookupAlg_.Process(ptr));
    CHK_RET(excutor->ExecuteOp(ptr));

    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::CollLookupSendKeys(HdcsLookupPipelineParams *pipelineParams)
{
    CHK_PTR_NULL(pipelineParams);
    s32 tag = pipelineParams->tag;
    s32 cubeIndex = pipelineParams->cubeIndex;

    HcclOperator<EmbeddingServiceParam> *ptr = lookupParamsMap_[tag][cubeIndex];
    CHK_PTR_NULL(ptr);

    ptr->param_.opType = EmbeddingServiceParam::OperatorType::REMOTE_LOOKUP_SEND_KEYS;
    ptr->param_.embeddingParam.cubeIndex = cubeIndex;

    CHK_RET(lookupAlg_.Process(ptr));

    OpExecutor *excutor = lookupOpExecutorMap_[tag][cubeIndex].get();
    CHK_PTR_NULL(excutor);
    CHK_RET(excutor->ExecuteOp(ptr));

    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::CollLookupRecvValues(HdcsLookupPipelineParams *pipelineParams)
{
    CHK_PTR_NULL(pipelineParams);
    s32 tag = pipelineParams->tag;
    s32 cubeIndex = pipelineParams->cubeIndex;

    HcclOperator<EmbeddingServiceParam> *ptr = lookupParamsMap_[tag][cubeIndex];
    CHK_PTR_NULL(ptr);

    ptr->param_.opType = EmbeddingServiceParam::OperatorType::REMOTE_LOOKUP_RECV_VALUES;
    ptr->param_.embeddingParam.cubeIndex = cubeIndex;

    CHK_RET(lookupAlg_.Process(ptr));

    OpExecutor *excutor = lookupOpExecutorMap_[tag][cubeIndex].get();
    CHK_PTR_NULL(excutor);
    CHK_RET(excutor->ExecuteOp(ptr));

    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::CollLookupRecoverValue(HdcsLookupPipelineParams *pipelineParams)
{
    CHK_PTR_NULL(pipelineParams);
    s32 tag = pipelineParams->tag;
    s32 cubeIndex = pipelineParams->cubeIndex;

    HcclOperator<EmbeddingServiceParam> *ptr = lookupParamsMap_[tag][cubeIndex];
    CHK_PTR_NULL(ptr);

    ptr->param_.opType = EmbeddingServiceParam::OperatorType::REMOTE_LOOKUP_RECOVER_VALUE;
    ptr->param_.embeddingParam.cubeIndex = cubeIndex;

    CHK_RET(lookupAlg_.Process(ptr));

    OpExecutor *excutor = lookupOpExecutorMap_[tag][cubeIndex].get();
    CHK_PTR_NULL(excutor);
    CHK_RET(excutor->ExecuteOp(ptr));

    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::CollLookupResetUniquesHandle(HdcsLookupPipelineParams *pipelineParams)
{
    CHK_PTR_NULL(pipelineParams);
    s32 tag = pipelineParams->tag;
    s32 cubeIndex = pipelineParams->cubeIndex;

    HcclOperator<EmbeddingServiceParam> *ptr = lookupParamsMap_[tag][cubeIndex];
    CHK_PTR_NULL(ptr);

    ptr->param_.opType = EmbeddingServiceParam::OperatorType::REMOTE_LOOKUP_RESET_UNIQUE_TASK;
    ptr->param_.embeddingParam.cubeIndex = cubeIndex;

    CHK_RET(lookupAlg_.Process(ptr));

    OpExecutor *excutor = lookupOpExecutorMap_[tag][cubeIndex].get();
    CHK_PTR_NULL(excutor);
    CHK_RET(excutor->ExecuteOp(ptr));

    return HCCL_SUCCESS;
}

// 流水时会用
HcclResult HdcsImpl::CollLookupWaitSendKeyFinish(HdcsLookupPipelineParams *pipelineParams)
{
    CHK_PTR_NULL(pipelineParams);
    s32 tag = pipelineParams->tag;
    HcclOperator<EmbeddingServiceParam> *ptr = embeddingRequestPool_.Alloc();
    CHK_PTR_NULL(ptr);

    ptr->param_.opType = EmbeddingServiceParam::OperatorType::REMOTE_LOOKUP_WAIT_SEND_KEY_FINISH;

    auto it = lookupOpExecutorMap_.Find(tag);
    if (it.second) {
        // 同一个tag内部不会并发
        auto &map = it.first->second;
        for (u32 cubeIndex = 0; cubeIndex < map.size(); ++cubeIndex) {
            CHK_RET(lookupAlg_.Process(ptr));

            OpExecutor *excutor = map[cubeIndex].get();
            CHK_PTR_NULL(excutor);
            HcclResult ret = excutor->ExecuteOp(ptr);
            if (ret != HCCL_SUCCESS) {
                HCCL_WARNING("ExecuteOp faild.");
                map.erase(cubeIndex);
            }
        }
    }

    embeddingRequestPool_.Free(ptr);
    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::RemoteUpdateKeyReduce(HdcsEmbeddingServiceParam *updateInfo)
{
    CHK_PTR_NULL(updateInfo);
    u64 keySize = updateInfo->keyMaxNum * SIZE_TABLE[updateInfo->keyType];
    u64 valueSize = updateInfo->keyMaxNum * updateInfo->valueItemSize;
    if (updateInfo->keyTransferMemSize < keySize || updateInfo->valueTransferMemSize < valueSize) {
        HCCL_ERROR("TransferMemSize not enough!!! keyTransferMemSize[%lld] valueTransferMemSize[%lld] keySize[%lld]"
            " valueSize[%lld]", updateInfo->keyTransferMemSize, updateInfo->valueTransferMemSize, keySize, valueSize);
        return HCCL_E_MEMORY;
    }

    auto allocRequest = [this] () -> HcclOperator<EmbeddingServiceParam>* {
        return embeddingRequestPool_.Alloc();
    };

    s32 tag = updateInfo->tag;

    pair<UniversalConcurrentMap<s32, HcclOperator<EmbeddingServiceParam>*>::Iterator, bool> it;
    EXECEPTION_CATCH(it = updateParamsMap_.EmplaceIfNotExist(tag, allocRequest), return HCCL_E_INTERNAL);
    if (it.second) {
        HCCL_DEBUG("updateParamsMap_ new insert, tag[%d]", tag);
    }

    HcclOperator<EmbeddingServiceParam> *opPtr = it.first->second;
    CHK_PTR_NULL(opPtr);

    EmbeddingServiceParam &param = opPtr->param_;
    param = *updateInfo;
    param.opType = EmbeddingServiceParam::OperatorType::REMOTE_UPDATE_KEY_REDUCE;
    param.embeddingParam.comm = comm_;

    if (param.pairedMode) {
        param.indices = updateInfo->indices;
        param.numUniqued = updateInfo->numUniqued;
        param.psSeg = updateInfo->psSeg;
        param.psSegNum = updateInfo->psSegNum;
    }

    // 在调用时捕获异常
    auto exeAlloc = [this] () -> unique_ptr<HdcsOpRemoteUpdateExecutor> {
        return make_unique<HdcsOpRemoteUpdateExecutor>();
    };

    pair<UniversalConcurrentMap<s32, unique_ptr<HdcsOpRemoteUpdateExecutor>>::Iterator, bool> exeIt;

    EXECEPTION_CATCH((exeIt = updateOpExecutorMap_.EmplaceIfNotExist(tag, exeAlloc)), return HCCL_E_INTERNAL);
    HdcsOpRemoteUpdateExecutor *excutor = exeIt.first->second.get();
    CHK_PTR_NULL(excutor);

    if (exeIt.second) {
        CHK_RET(excutor->Init(psRankId_));
        HCCL_DEBUG("updateOpExecutorMap_ new insert, tag[%d]", tag);
    }

    CHK_RET(updateAlg_.Process(opPtr));
    CHK_RET(excutor->ExecuteOp(opPtr));
    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::RemoteUpdateReduceSum(HdcsEmbeddingServiceParam *updateInfo)
{
    CHK_PTR_NULL(updateInfo);
    s32 tag = updateInfo->tag;

    HcclOperator<EmbeddingServiceParam> *ptr = updateParamsMap_[tag];
    CHK_PTR_NULL(ptr);

    EmbeddingServiceParam &param = ptr->param_;
    param.opType = EmbeddingServiceParam::OperatorType::REMOTE_UPDATE_REDUCE_SUM;
    param.pairedMode = updateInfo->pairedMode;
    param.haveRdmaConn = updateInfo->haveRdmaConn;

    CHK_RET(updateAlg_.Process(ptr));

    HdcsOpRemoteUpdateExecutor *excutor = updateOpExecutorMap_[tag].get();
    CHK_PTR_NULL(excutor);
    CHK_RET(excutor->ExecuteOp(ptr));

    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::RemoteUpdateSendRequest(HdcsEmbeddingServiceParam *updateInfo)
{
    CHK_PTR_NULL(updateInfo);

    s32 tag = updateInfo->tag;
    HcclOperator<EmbeddingServiceParam> *ptr = updateParamsMap_[tag];
    CHK_PTR_NULL(ptr);

    EmbeddingServiceParam &param = ptr->param_;
    param.opType = EmbeddingServiceParam::OperatorType::REMOTE_UPDATE_SEND_REQUEST;
    param.haveRdmaConn = updateInfo->haveRdmaConn;
    param.tableId = updateInfo->tableId;
    param.tableIdAddr = updateInfo->tableIdAddr;
    param.globalStepAddr = updateInfo->globalStepAddr;

    CHK_RET(updateAlg_.Process(ptr));

    HdcsOpRemoteUpdateExecutor *excutor = updateOpExecutorMap_[tag].get();
    CHK_PTR_NULL(excutor);
    CHK_RET(excutor->ExecuteOp(ptr));

    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::RemoteUpdateResetUniqueHandle(HdcsEmbeddingServiceParam *updateInfo)
{
    CHK_PTR_NULL(updateInfo);
    s32 tag = updateInfo->tag;

    HcclOperator<EmbeddingServiceParam> *ptr = updateParamsMap_[tag];
    CHK_PTR_NULL(ptr);

    EmbeddingServiceParam &param = ptr->param_;
    param.opType = EmbeddingServiceParam::OperatorType::REMOTE_UPDATE_RESET_UNIQUE_HANDLE;
    param.pairedMode = updateInfo->pairedMode;
    param.haveRdmaConn = updateInfo->haveRdmaConn;

    CHK_RET(updateAlg_.Process(ptr));

    HdcsOpRemoteUpdateExecutor *excutor = updateOpExecutorMap_[tag].get();
    CHK_PTR_NULL(excutor);
    CHK_RET(excutor->ExecuteOp(ptr));

    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::RemoteUpdateRecvResponse(HdcsEmbeddingServiceParam *updateInfo)
{
    CHK_PTR_NULL(updateInfo);
    s32 tag = updateInfo->tag;

    HcclOperator<EmbeddingServiceParam> *ptr = updateParamsMap_[tag];
    CHK_PTR_NULL(ptr);

    EmbeddingServiceParam &param = ptr->param_;
    param.opType = EmbeddingServiceParam::OperatorType::REMOTE_UPDATE_RECV_RESPONSE;
    param.haveRdmaConn = updateInfo->haveRdmaConn;

    CHK_RET(updateAlg_.Process(ptr));

    HdcsOpRemoteUpdateExecutor *excutor = updateOpExecutorMap_[tag].get();
    CHK_PTR_NULL(excutor);
    CHK_RET(excutor->ExecuteOp(ptr));

    return HCCL_SUCCESS;
}

HcclResult HdcsImpl::ChkServiceCancel(HdcsEmbeddingServiceParam *inferenceParams)
{
    CHK_PTR_NULL(inferenceParams);
    CHK_PTR_NULL(inferenceParams->errorMsg);
    if (*(static_cast<u32 *>(inferenceParams->errorMsg)) == SERVICE_CANCEL_SIGNAL) {
        HCCL_ERROR("[HdcsImpl]hdcs ChkServiceCancel errorMsg[%u]", *(static_cast<u32 *>(inferenceParams->errorMsg)));
        return HCCL_E_INTERNAL;
    }

    return HCCL_SUCCESS;
}
}
